package cn.edu.flink.scala.tutorial.state.keyedstate

import cn.edu.flink.scala.tutorial.source.{ClickSource, Event}
import org.apache.flink.api.common.state.{StateTtlConfig, ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object TtlTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()


    val clickStream = env.addSource(new ClickSource).assignAscendingTimestamps(_.timestamp)

    clickStream.keyBy(_.user)
      .process(new KeyedProcessFunction[String, Event, String] {
        var countState: ValueState[Long] = _

        override def open(parameters: Configuration): Unit = {
          val countStateDescriptor = new ValueStateDescriptor[Long]("count state", classOf[Long])
          // 生命周期
          val ttlConfig: StateTtlConfig = StateTtlConfig
            .newBuilder(Time.seconds(1))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
            .build()

          countStateDescriptor.enableTimeToLive(ttlConfig)

          countState = getRuntimeContext.getState(countStateDescriptor)
        }


        override def processElement(value: Event, ctx: KeyedProcessFunction[String, Event, String]#Context, out: Collector[String]): Unit = {
          countState.update(countState.value() + 1L)

          out.collect(value.user + ": " + countState.value())
        }
      })
      .print()

    env.execute("TtlTest")
  }
}
